/*
 * Copyright (c) 2018-2023 北京华恒盛世科技有限公司
 * QUICKPOOL is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *     http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
*/

#include "channel.h"
#include "common.h"
#include "api.h"

extern char **environ;

/**
 * @brief       API: call remote node to start my task.
 *
 * @param[in]   cnode        #1: remote node and my task
 * @param[in]   port         #2: port of remote node
 * @param[in]   timeout      #3: connection time out
 * @param[in]   options      #4: options for starting task
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed, connect fail, system call fail, xdr error
 *
 * @note        A task will be started on remote node.
 */
int
execute_task(struct conn_node *cnode, int port, int timeout, int options)
{
    int numEnv, numArgc;

    for (numEnv=0; environ[numEnv]; numEnv++) ;

    for (numArgc=0; cnode->task_argv[numArgc]; numArgc++) ;

    int fd;
    fd = gf_connect(cnode->node_name, port, timeout);
    if (0 > fd) {
        if (-1 == fd) {
            log_message(ERR, NOCLASS, "%s: cannot get my host name [%s] from OS.",
                        __func__, cnode->node_name);
        } else if (-2 == fd) {
            if (0 == errno) {
                log_message(ERR, NOCLASS, "%s: connect to host[%s] failed.",
                            __func__, cnode->node_name);
            } else {
                log_message(ERR, NOCLASS, "%s: connect to host[%s] failed: %m.",
                            __func__, cnode->node_name);
            }
        } else if (-3 == fd) {
            log_message(ERR, NOCLASS, "%s: connect to host[%s] timeout.",
                        __func__, cnode->node_name);
        }

	return -1;
    }

    if (gf_new_conn(fd) < 0) {
        /* cannot find free fd, wait a while */
        log_message(ERR, NOCLASS, "%s: no available fd now, wait for a while.", __func__);
        close(fd);
        return -1;
    }

    cnode->conn_fd = fd;

    struct sockaddr_in sin;
    sigset_t newMask;
    sigset_t oldMask;
    if (gf_block_allsignals(&newMask, &oldMask) < 0) {

        log_message(ERR, NOCLASS, "%s: gf_block_allsignals failed when execute task on remote node <%s> due to %m.",
                    __func__, cnode->node_name);

        gf_chan_close(fd);
        cnode->conn_fd = -1;

        return -1;
    }

    socklen_t len;
    if (getpeername(fd, (struct sockaddr *)&sin, &len) < 0) {
        log_message(ERR, NOCLASS, "%s: getpeername failed when execute task on remote node <%s> due to %m.",
                    __func__, cnode->node_name);

        gf_chan_close(fd);
        cnode->conn_fd = -1;
        quickpool_errno = ERROR_SYS_CALL;
        gf_unblock_signals(&oldMask);

        return -1;
    }

    struct request taskRequest;
    memset(&taskRequest, 0, sizeof(struct request));

    unsigned int n = 0;
    taskRequest.keyvalues = (struct keyvalue *)calloc((15+numArgc+numEnv), sizeof(struct keyvalue));
    if (NULL == taskRequest.keyvalues) {

        quickpool_errno = ERROR_MEMORY;
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc",
                    (15+numArgc+numEnv)*sizeof(struct keyvalue));
        gf_chan_close(fd);
        cnode->conn_fd = -1;
        gf_unblock_signals(&oldMask);

        return -1;
    }

    taskRequest.keyvalues[n].key = REQUEST_KEY_OPTIONS;
    taskRequest.keyvalues[n].value = (char *)calloc(12, sizeof(char));
    if (NULL == taskRequest.keyvalues[n].value) {

        quickpool_errno = ERROR_MEMORY;
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        gf_chan_close(fd);
        cnode->conn_fd = -1;
        gf_unblock_signals(&oldMask);

        return -1;
    }
    snprintf(taskRequest.keyvalues[n].value, 12, "%d", options);
    n ++;

    taskRequest.keyvalues[n].key = REQUEST_KEY_JOBIDS;
    taskRequest.keyvalues[n].value = (char *)calloc(32, sizeof(char));
    if (NULL == taskRequest.keyvalues[n].value) {

        quickpool_errno = ERROR_MEMORY;
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 32);
        gf_chan_close(fd);
        cnode->conn_fd = -1;
        gf_unblock_signals(&oldMask);

        return -1;
    }
    snprintf(taskRequest.keyvalues[n].value, 32, "%lld", cnode->jobid);
    n ++;

    taskRequest.keyvalues[n].key = REQUEST_KEY_NUM_TASK;
    taskRequest.keyvalues[n].value = (char *)calloc(12, sizeof(char));
    if (NULL == taskRequest.keyvalues[n].value) {

        quickpool_errno = ERROR_MEMORY;
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        gf_chan_close(fd);
        cnode->conn_fd = -1;
        gf_unblock_signals(&oldMask);

        return -1;
    }
    snprintf(taskRequest.keyvalues[n].value, 12, "%d", cnode->num_task);
    n ++;

    taskRequest.keyvalues[n].key = REQUEST_KEY_NODE_RANK;
    taskRequest.keyvalues[n].value = (char *)calloc(12, sizeof(char));
    if (NULL == taskRequest.keyvalues[n].value) {

        quickpool_errno = ERROR_MEMORY;
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        gf_chan_close(fd);
        cnode->conn_fd = -1;
        gf_unblock_signals(&oldMask);

        return -1;
    }
    snprintf(taskRequest.keyvalues[n].value, 12, "%d", cnode->node_rank);
    n ++;

    if (NULL != cnode->execute_user) {
        taskRequest.keyvalues[n].key = REQUEST_KEY_USERNAME;
        taskRequest.keyvalues[n].value = copy_string(cnode->execute_user);
        n ++;
    }

    if (NULL != cnode->cwd) {
        taskRequest.keyvalues[n].key = REQUEST_KEY_EXECUTE_CWD;
        taskRequest.keyvalues[n].value = copy_string(cnode->cwd);
        n ++;
    }

    taskRequest.keyvalues[n].key = REQUEST_KEY_HOSTNAME;
    taskRequest.keyvalues[n].value = copy_string(cnode->firstnode_name);
    n ++;

    if (0 < cnode->firstnode_port) {
        taskRequest.keyvalues[n].key = REQUEST_KEY_FIRSTNODE_PORT;
        taskRequest.keyvalues[n].value = (char *)calloc(12, sizeof(char));
        if (NULL == taskRequest.keyvalues[n].value) {

            quickpool_errno = ERROR_MEMORY;
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            gf_chan_close(fd);
            cnode->conn_fd = -1;
            gf_unblock_signals(&oldMask);

            return -1;
        }
        snprintf(taskRequest.keyvalues[n].value, 12, "%d", cnode->firstnode_port);
        n ++;
    }

    if (NULL != cnode->script) {
        taskRequest.keyvalues[n].key = REQUEST_KEY_JOB_FILE;
        taskRequest.keyvalues[n].value = copy_string(cnode->script);
        n ++;
    }

    taskRequest.keyvalues[n].key = REQUEST_KEY_TASK_ARGC;
    taskRequest.keyvalues[n].value = (char *)calloc(12, sizeof(char));
    if (NULL == taskRequest.keyvalues[n].value) {

        quickpool_errno = ERROR_MEMORY;
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        gf_chan_close(fd);
        cnode->conn_fd = -1;
        gf_unblock_signals(&oldMask);

        return -1;
    }
    snprintf(taskRequest.keyvalues[n].value, 12, "%d", numArgc);
    n ++;

    numArgc = 0;
    while (NULL != cnode->task_argv[numArgc]) {
        taskRequest.keyvalues[n].key = REQUEST_KEY_TASK_ARGV;
        taskRequest.keyvalues[n].value = copy_string(cnode->task_argv[numArgc]);
        numArgc ++;
        n ++;
    }
    taskRequest.number = n;

    struct auth auth;
    gf_set_auth(&auth, commandParams[AUTH_KIND].string_value);

    unsigned int size; // for calculating the size of the request.
    size = gf_xdrsize_request(&taskRequest);
    size += gf_xdrsize_auth(&auth);
    size += HEADER_LEN;

    char *request;
    request = (char *)calloc(size, sizeof(char));
    if (NULL == request) {

        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", size);
        gf_chan_close(fd);
        cnode->conn_fd = -1;
        gf_xdr_free(gf_xdr_auth, (void *)&auth);
        gf_xdr_free(gf_xdr_request, &taskRequest);
        quickpool_errno = ERROR_MEMORY;
        gf_unblock_signals(&oldMask);

        return -1;
    }

    XDR xdrs;
    xdrmem_create(&xdrs, request, size, XDR_ENCODE);

    struct header tohdr;
    gf_init_header(&tohdr);
    tohdr.reqcode = CMD_START_TASK;

    if (!gf_xdr_message(&xdrs, &taskRequest, &tohdr, &auth, gf_xdr_request, NULL)) {

        // fail to encode request.
        log_message(ERR, NOCLASS, STR_FUNC_FAIL, __func__, "gf_xdr_message");
        gf_chan_close(fd);
        cnode->conn_fd = -1;
        xdr_destroy(&xdrs);
        FREEUP(request);
        gf_xdr_free(gf_xdr_auth, (void *)&auth);
        gf_xdr_free(gf_xdr_request, &taskRequest);
        quickpool_errno = ERROR_XDR;
        gf_unblock_signals(&oldMask);

        return -1;
    } 

    char strJobId[32];

    unsigned int requestSize;
    requestSize = XDR_GETPOS(&xdrs);

    unsigned int sendLen;
    sendLen = gf_chan_write(fd, request, requestSize, timeout);
    if (sendLen != requestSize) {

        log_message(ERR, NOCLASS, "%s: Failed to start task on node <%s> for job <%s>",
                    __func__, cnode->node_name, jobid_2string(cnode->jobid, strJobId));
        gf_chan_close(fd);
        cnode->conn_fd = -1;
        gf_unblock_signals(&oldMask);

        return (-1);
    }

    struct header outhdr;
    gf_init_header(&outhdr);

    char *reply=NULL;
    int cc;
    cc = gf_read_reply(fd, g_recvtimeout, &outhdr, &reply);
    if (cc < 0) {

        gf_chan_close(fd);
        cnode->conn_fd = -1;
        gf_unblock_signals(&oldMask);

        return -1;
    }

    struct epoll_event ev;
    memset(&ev, 0, sizeof(struct epoll_event));
    ev.data.fd = fd;
    ev.events = EPOLLIN;
    epoll_ctl(gf_get_epfd(), EPOLL_CTL_ADD, fd, &ev);

    gf_unblock_signals(&oldMask);

    gf_xdr_free(gf_xdr_request, &taskRequest);

    return (0);
} // end function execute_task

/**
 * @brief       API: send signal to task on remote node.
 *
 * @param[in]   cnode        #1: remote node and my task
 * @param[in]   signal       #2: signal value to be sent
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed, connect fail, system call fail, xdr error
 *
 * @note        A task will be signaled on remote node.
 */
int
send_signal_2task(struct conn_node *cnode, int signal)
{
    struct request sigJobReq;
    memset(&sigJobReq, 0, sizeof(struct request));

    unsigned int num = 2; // jobid, signal
    sigJobReq.keyvalues = (key_value_t *)calloc(num, sizeof(key_value_t));
    if (NULL == sigJobReq.keyvalues) {

        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc",
                    num*sizeof(struct keyvalue));
        return ERROR_MEMORY;
    }

    num = 0;
    sigJobReq.keyvalues[num].key = REQUEST_KEY_JOBIDS;
    sigJobReq.keyvalues[num].value = (char *)calloc(32, sizeof(char));
    if (NULL == sigJobReq.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 32);
        return ERROR_MEMORY;
    }
    snprintf(sigJobReq.keyvalues[num].value, 32, "%lld", cnode->jobid);
    num ++;

    sigJobReq.keyvalues[num].key = REQUEST_KEY_SIGNAL;
    sigJobReq.keyvalues[num].value = (char *)calloc(12, sizeof(char));
    if (NULL == sigJobReq.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        return ERROR_MEMORY;
    }
    snprintf(sigJobReq.keyvalues[num].value, 12, "%d", signal);
    num ++;

    sigJobReq.number = num;

    struct auth auth;
    gf_set_auth(&auth, commandParams[AUTH_KIND].string_value);

    unsigned int size; // for calculating the size of the request.
    size = gf_xdrsize_request(&sigJobReq);
    size += HEADER_LEN;

    char *request;
    request = (char *)calloc(size, sizeof(char));
    if (NULL == request) {

        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", size);
        gf_xdr_free(gf_xdr_auth, (void *)&auth);
        gf_xdr_free(gf_xdr_request, &sigJobReq);

        return ERROR_MEMORY;
    }

    XDR xdrs;
    xdrmem_create(&xdrs, request, size, XDR_ENCODE);

    struct header hdr;
    gf_init_header(&hdr);
    hdr.reqcode = JM_SIGNAL_TASK;

    if (!gf_xdr_message(&xdrs, (void *)&sigJobReq, &hdr, &auth, gf_xdr_request, NULL)) {

        log_message(ERR, NOCLASS, STR_FUNC_FAIL, __func__, "gf_xdr_message");

        xdr_destroy(&xdrs);
        FREEUP(request);
        gf_xdr_free(gf_xdr_auth, (void *)&auth);
        gf_xdr_free(gf_xdr_request, &sigJobReq);

        return ERROR_XDR;
    }

    int fd = -1, ret;
    ret = gf_callslave(cnode->node_name, g_jm_port, g_conntimeout, g_recvtimeout,
                       XDR_GETPOS(&xdrs), request, NULL, NULL, &fd);

    xdr_destroy(&xdrs);
    FREEUP(request);
    gf_xdr_free(gf_xdr_auth, (void *)&auth);
    gf_xdr_free(gf_xdr_request, &sigJobReq);

    if (0 > ret) {

        if (ret == -2) {
            return (ERROR_DAEMON_DOWN);
        }

        return (ERROR_COMMUNICATE);
    }

    return 0;
} // end function send_signal_2task
